package com.atguigu.day03;

import com.atguigu.util.IntegerSource;
import com.atguigu.util.IntegerStatistic;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class Example8 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntegerSource())
                .keyBy(r -> "int")
                .process(new KeyedProcessFunction<String, Integer, IntegerStatistic>() {
                    private ValueState<IntegerStatistic> accumulator;
                    // 标志位，如果为null,那么说明不存在定时器
                    private ValueState<Boolean> isTimerExisted;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        accumulator = getRuntimeContext().getState(
                                new ValueStateDescriptor<>(
                                        "acc",
                                        Types.POJO(IntegerStatistic.class)
                                )
                        );
                        isTimerExisted = getRuntimeContext().getState(
                                new ValueStateDescriptor<>(
                                        "is-timer-existed",
                                        Types.BOOLEAN
                                )
                        );
                    }

                    @Override
                    public void processElement(Integer in, Context context, Collector<IntegerStatistic> out) throws Exception {
                        if (accumulator.value() == null) {
                            accumulator.update(new IntegerStatistic(in, in, in, 1, in));
                        } else {
                            var oldAcc = accumulator.value();
                            var newAcc = new IntegerStatistic(
                                    Math.max(oldAcc.max, in),
                                    Math.min(oldAcc.min, in),
                                    oldAcc.sum + in,
                                    oldAcc.count + 1,
                                    (oldAcc.sum + in) / (oldAcc.count + 1)
                            );
                            accumulator.update(newAcc);
                        }

                        // 如果数据到来时，此时不存在发送统计结果的定时器
                        if (isTimerExisted.value() == null) {
                            context.timerService().registerProcessingTimeTimer(
                                    context.timerService().currentProcessingTime() + 10 * 1000L
                            );
                            // 因为注册了定时器，此时已经存在定时器了
                            // 将标志位置为非空
                            isTimerExisted.update(true);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<IntegerStatistic> out) throws Exception {
                        // 定时器的作用是发送统计结果
                        out.collect(accumulator.value());
                        // 当onTimer触发时，定时器时间戳已经从定时器时间戳队列中删除了，所以标志为置为null
                        isTimerExisted.clear();
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
